package ltd.hyangyang.rpc.core.remoting.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import ltd.hyangyang.rpc.common.bean.Request;
import ltd.hyangyang.rpc.common.bean.Response;
import ltd.hyangyang.rpc.common.constant.RemoteMsgConstant;
import ltd.hyangyang.rpc.common.utils.AddrUtils;
import ltd.hyangyang.rpc.core.serialize.RpcDecoder;
import ltd.hyangyang.rpc.core.serialize.RpcEncoder;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

public class RpcClientNetty {

    private EventLoopGroup group = new NioEventLoopGroup();

    static ConcurrentHashMap<String, Channel> generalChannel = new ConcurrentHashMap<>();

    Logger logger =Logger.getGlobal();

    String addr;

    Channel channel;

    public RpcClientNetty(String addr) throws InterruptedException {
        this.addr = addr;
        channel = generalChannel.get(addr);
        if (channel!=null&&channel.isActive()){
            return;
        }
        Bootstrap bootstrap = new Bootstrap();
        ChannelFuture channelFuture = bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<NioSocketChannel>() {
                    @Override
                    protected void initChannel(NioSocketChannel nioSocketChannel) throws Exception {
                        //
                        ChannelPipeline pipeline = nioSocketChannel.pipeline();
                        pipeline.addLast(new RpcEncoder());
                        pipeline.addLast(new RpcDecoder(Response.class));
//                        pipeline.addLast(new IdleStateHandler(100,100,3, TimeUnit.SECONDS));
                        pipeline.addLast(new RpcClientHandler());
                    }
                })
                .connect(AddrUtils.getInetSocketAddress(addr))
                .sync();
        channelFuture.addListener(future -> {
            if (future.isSuccess()){
                //如果连接成功，调用
                logger.info("连接服务成功");
            }
            if (future.isCancellable()){
                //如果连接识别，调用
                generalChannel.remove(addr);
            }
        });
        channel = channelFuture.channel();
        generalChannel.putIfAbsent(addr,channel);
    }

    public CompletableFuture<Response> sendMessage(Request request) {
        CompletableFuture<Response> progressFuture = UnfinishChannel.getProgressFuture(request.getRequestId());
        channel.writeAndFlush(request).addListener(future -> {
           if (future.isSuccess()){
               logger.info("将请求内容成功发送到服务端");
           }

        });

        return progressFuture;
    }


    public void sendPing(Channel channel) {
        Request ping = Request.builder().msgType(RemoteMsgConstant.PING).build();
        channel.writeAndFlush(ping).addListener(future -> {
           if (!future.isSuccess()){
               //失败
               channel.close();
           }
        });
    }
}
